package com.atguigu.realtime.app.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author chenlongStart
 * @create 2021-06-26 14:51
 * @desc
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private Connection connection;
    private ValueState<Boolean> createTableState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 0. 建立到Phoenix的连接. 使用标准的jdbc连接就可以了
        // 1. 加载驱动(常见的数据库可以不用加载, java会根据url自动的加载, 有些数据库比较加载, 比如: Phoenix)
        Class.forName(Constant.PHOENIX_DRIVER);
        // 2. 获取连接对象
        connection = DriverManager.getConnection(Constant.PHOENIX_URL);
        createTableState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("createTableState", Boolean.class));

    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
        // 1. 检测表是否存在, 如果不存在需要先建表
        checkTable(value);
        // 2.  把这条数据写入到Phoenix中
        write2Hbase(value);
        // 3. dwm层需要的: 更新redis中的缓存
        // 如果检测到这次是update操作, 则需要去缓存中更新对应的维度数据或者删除对应的数据
        delCache(value);
    }

    private void delCache(Tuple2<JSONObject, TableProcess> value) {
        String key = value.f1.getSinkTable().toUpperCase() + ":" + value.f0.get("id");
        Jedis jedis = RedisUtil.getRedisClient();
        jedis.select(1);
        jedis.del(key);// 如果key不存在, 会有啥影响?  没有影响. 所以不用提前判断key是否存在
        jedis.close();// 归还给连接池
    }

    @Override
    public void close() throws Exception {
        if (connection != null) {
            connection.close();
        }
    }

    private void write2Hbase(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        StringBuilder sql = new StringBuilder();
        sql
                .append("upsert into ")
                .append(tp.getSinkTable())
                .append("(")
                .append(tp.getSinkColumns())
                .append(")values(")
                .append(tp.getSinkColumns().replaceAll("[^,]+","?"))
                .append(")");
        PreparedStatement ps = connection.prepareStatement(sql.toString());
        System.out.println("写入语句："+sql.toString());
        String[] columns = tp.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String v = data.get(columns[i]) == null ? "" : data.get(columns[i]).toString();
            ps.setString(i+1,v);
        }
        ps.execute();
        connection.commit();
        ps.close();
    }

    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
//        System.out.println("PhoenixSink.checkTable");


        if (createTableState.value() == null) {
            TableProcess tp = value.f1;
            StringBuilder sql = new StringBuilder();
            sql
                    .append("create table if not exists ")
                    .append(tp.getSinkTable())
                    .append("(")
                    .append(tp.getSinkColumns().replaceAll(","," varchar,"))
                    .append(" varchar, constraint ps primary key(")
                    .append(tp.getSinkPk()==null?"id":tp.getSinkPk())
                    .append(") )")
                    .append(tp.getSinkExtend()==null?"":tp.getSinkExtend());
            System.out.println("建表语句："+sql.toString());
            PreparedStatement ps = connection.prepareStatement(sql.toString());
            ps.execute();
            connection.commit();
            ps.close();
            createTableState.update(true);
        }
    }
}
